package prictice.transaction1;

import org.apache.storm.coordination.BatchOutputCollector;
import org.apache.storm.task.TopologyContext;
import org.apache.storm.topology.OutputFieldsDeclarer;
import org.apache.storm.topology.base.BaseTransactionalBolt;
import org.apache.storm.transactional.TransactionAttempt;
import org.apache.storm.tuple.Fields;
import org.apache.storm.tuple.Tuple;
import org.apache.storm.tuple.Values;

import java.util.Map;

/**
 * @Package prictice.transaction1
 * @Description:
 * @Author elwyn
 * @Date 2017/9/16 20:13
 * @Email elonyong@163.com
 */
public class MyTranscationBolt extends BaseTransactionalBolt {
    private static final long serialVersionUID = -2751063873826429097L;
    BatchOutputCollector batchOutputCollector;
    TransactionAttempt value;

    @Override
    public void prepare(Map map, TopologyContext topologyContext, BatchOutputCollector batchOutputCollector, TransactionAttempt transactionAttempt) {
        this.batchOutputCollector = batchOutputCollector;
        System.err.println("MyTranscationBolt prepare " + transactionAttempt.getTransactionId() + "  attemptId" + transactionAttempt.getAttemptId());
    }

    Integer count = 0;

    @Override
    public void execute(Tuple tuple) {
        value = (TransactionAttempt) tuple.getValue(0);
        System.err.println("MyTranscationBolt TranscationAttempt " + value.getTransactionId() + "  attemptId" + value.getAttemptId());
        String log = tuple.getString(1);
        if (log != null && log.length() > 0) {
            count++;
        }
    }

    @Override
    public void finishBatch() {
        batchOutputCollector.emit(new Values(value, count));

    }

    @Override
    public void declareOutputFields(OutputFieldsDeclarer outputFieldsDeclarer) {
        outputFieldsDeclarer.declare(new Fields("tx", "count"));
    }
}
